package com.framework.flink;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple0;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.plan.nodes.datastream.TimeCharacteristic;

/**
 * 描述：
 * 文件读取
 * @author xianggj
 * @Date 2021/10/19 15:00
 **/
public class FlinkTestFileSource {

//    public static void main(String[] args) {
//        try {
//            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//            env.enableCheckpointing(5000); // 非常关键，一定要设置启动检查点！！
//            env.readFile();
////            env.getConfig().setGlobalJobParameters(loadConfig("config.properties"));
////            DataStream<String> text = env.socketTextStream("localhost", 9090, "\n");
////            DataStream stream = text.flatMap(new LineMapper());
////            //stream.addSink(new SinkTest());
////            // 将结果打印到控制台，注意这里使用的是单线程打印，而非多线程
////            //env.setParallelism(1);
////            env.execute("WordCount from Kafka data");
//        } catch (Exception e) {
//            System.out.println(e);
//        }
//
//    }
public static void main(String[] args) throws Exception {
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    csvfile(env);
}

    public static void csvfile(ExecutionEnvironment env) throws Exception {
        String filePath = "D:\\test.csv";
        DataSet<Tuple4<Integer, String, String, Integer>> csvInput =
                env.readCsvFile(filePath)
                .fieldDelimiter(",")
                //忽略第一行
                .ignoreFirstLine()
                //过滤字段 true表示选择某列 false表示忽略某列
                .includeFields(true,true,true,true)
                .types(Integer.class, String.class, String.class, Integer.class);
//                .pojoType(newPerson.class,"name","address");
        csvInput.print();
    }

}
